package com.superbytecode.cloud.rocketmq.consumer.provider;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.OrderConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.superbytecode.cloud.common.model.mq.MQTopic;
import com.superbytecode.cloud.common.utils.VerifyUtil;
import com.superbytecode.cloud.rocketmq.consumer.Consumer;
import com.superbytecode.cloud.rocketmq.configure.AliMQConfiguration;
import com.superbytecode.cloud.rocketmq.configure.MQProviderEnum;
import com.superbytecode.cloud.rocketmq.consumer.MQConsumeProvider;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.springframework.core.ResolvableType;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;

/**
 * All rights Reserved, Designed By www.super-master.com
 *
 * @ProjectName: beacon-cloud
 * @Package: com.superbytecode.cloud.rocketmq.consumer
 * @ClassName: AliRocketMQConsumeContainer
 * @Description: []
 * <p> </p>
 * @Author: [Lance Ting]
 * @Date: 2024/11/1 17:53
 * @Version: V1.0
 * @Copyright: 2024 www.super-master.com Inc. All rights reserved.
 * TODO: 注意,本文件Lance Ting所作,如果转载或使用请标明具体出处!
 **/
@Slf4j
public class AliMQConsumeProvider implements MQConsumeProvider {
    private final ObjectMapper objectMapper;
    private final AliMQConfiguration configuration;

    public AliMQConsumeProvider(AliMQConfiguration configuration, JavaTimeModule module) {
        this.configuration = configuration;
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(module);
    }

    /**
     * 注册容器
     *
     * @param targetClass 目标类
     * @param beanName    bean名称
     * @param beanObj     消费者的bean
     * @param topic       topic名字
     * @param group       消费组id
     * @param consumer    消费者配置
     */
    @Override
    public void register(Class<?> targetClass, String beanName, Object beanObj, String topic, String group, Consumer consumer) {
        //配置文件
        Properties properties = configuration.initAliMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, group);
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        String messageModeCN = "";
        switch (consumer.messageModel()) {
            case BROADCASTING:
                messageModeCN = MessageModel.BROADCASTING.getModeCN();
                break;
            case CLUSTERING:
                messageModeCN = MessageModel.CLUSTERING.getModeCN();
                break;
        }
        if (VerifyUtil.isNotEmpty(messageModeCN)) {
            properties.setProperty(PropertyKeyConst.MessageModel, messageModeCN);
        }
        @SuppressWarnings("unchecked")//
        MQTopic<Object> mqTopic = (MQTopic<Object>) beanObj;
        ResolvableType[] resolvableInterfaces = ResolvableType.forClass(targetClass).getInterfaces();
        ResolvableType resolvableType = resolvableInterfaces[0].getInterfaces()[0].getGeneric(0);
        Class<?> messageType = resolvableType.getRawClass();
        //订阅关系
        String tag = consumer.tag();
        Subscription subscription = new Subscription();
        subscription.setTopic(topic);
        subscription.setExpression(tag);
        //订阅多个topic如上面设置
        ConsumeMode consumeMode = consumer.consumeMode();
        switch (consumeMode) {
            case CONCURRENTLY:
                consumer(mqTopic, messageType, subscription, properties);
                break;
            case ORDERLY:
                orderConsumer(mqTopic, messageType, subscription, properties);
                break;
            default:
                break;
        }
        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(), UUID.randomUUID());
        log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}, topic: {}, groupId: {}", //
                beanName, containerBeanName, topic, group);
    }

    private void consumer(MQTopic<Object> topic, Class<?> messageType, Subscription subscription, Properties properties) {
        MessageListener messageListener = (message, context) -> {
            log.info("consumer {}", message);
            byte[] body = message.getBody();
            //
            Object body1;
            try {
                // 反序列化 如果报错了，就直接跳过
                body1 = objectMapper.readValue(message.getBody(), messageType);
            } catch (IOException e) {
                log.error("aliMQ consumer消费失败, {}", message, e);
                return Action.CommitMessage;
            }
            try {
                topic.consume(body);
            } catch (Exception e) {
                return Action.ReconsumeLater;
            }
            return Action.CommitMessage;
        };
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>(1);
        subscriptionTable.put(subscription, messageListener);
        ConsumerBean consumerBean = new ConsumerBean();
        consumerBean.setProperties(properties);
        consumerBean.setSubscriptionTable(subscriptionTable);
        consumerBean.start();
    }

    private void orderConsumer(MQTopic<Object> topic, Class<?> messageType, Subscription subscription, Properties properties) {
        MessageOrderListener messageListener = (message, context) -> {
            log.info("consumer {}", message);
            message.getBody();
            Object body;
            try {
                //反序列化 如果报错了，就直接跳过
                body = objectMapper.readValue(message.getBody(), messageType);
            } catch (IOException e) {
                log.error("alimq 消费失败, {}", message, e);
                return OrderAction.Success;
            }
            try {
                topic.consume(body);
            } catch (Exception e) {
                return OrderAction.Suspend;
            }

            return OrderAction.Success;
        };
        Map<Subscription, MessageOrderListener> subscriptionTable = new HashMap<>(1);
        subscriptionTable.put(subscription, messageListener);
        OrderConsumerBean consumerBean = new OrderConsumerBean();
        consumerBean.setProperties(properties);
        consumerBean.setSubscriptionTable(subscriptionTable);
        consumerBean.start();
    }

    /**
     * QM服务提供者
     *
     * @return 阿里MQ|开源MQ
     */
    @Override
    public MQProviderEnum provider() {
        return MQProviderEnum.ALI;
    }


    //


}